package com.xpc.rpc.register.zookeeper;

import com.xpc.rpc.loadbalancer.api.ServiceLoadBalancerApi;
import com.xpc.rpc.protocol.meta.ServiceMeta;
import com.xpc.rpc.register.common.RegisterService;
import com.xpc.rpc.register.common.config.RegisterConfig;
import com.xpc.rpc.spi.loader.ExtensionLoader;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.x.discovery.ServiceDiscovery;
import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;
import org.apache.curator.x.discovery.ServiceInstance;
import org.apache.curator.x.discovery.details.JsonInstanceSerializer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collection;
import java.util.List;

/**
 * zookeeper注册中心功能实现
 */
public class ZookeeperRegisterServiceImpl implements RegisterService {

    private static final Logger LOGGER = LoggerFactory.getLogger(ZookeeperRegisterServiceImpl.class);

    private static final int BASE_SLEEP_TIME_MS = 1000;

    private static final int MAX_RETRIES = 3;

    private static final String ZK_BASE_PATH = "/xpc_rpc";

    private ServiceDiscovery<ServiceMeta> serviceDiscovery;


    @Override
    public void register(ServiceMeta serviceMeta) throws Exception {
        ServiceInstance<ServiceMeta> serviceInstance = ServiceInstance
                .<ServiceMeta>builder()
                .name(serviceMeta.getInterfaceName())
                .address(serviceMeta.getServerAddress())
                .port(serviceMeta.getServerPort())
                .payload(serviceMeta)
                .build();
        serviceDiscovery.registerService(serviceInstance);
    }

    @Override
    public ServiceMeta discovery(String interfaceName,String loadbalancerType) throws Exception {
        Collection<ServiceInstance<ServiceMeta>> serviceInstances = serviceDiscovery.queryForInstances(interfaceName);
        ServiceLoadBalancerApi serviceLoadBalancerApi = ExtensionLoader.getExtension(ServiceLoadBalancerApi.class, loadbalancerType);
        ServiceInstance<ServiceMeta> serviceInstance = (ServiceInstance<ServiceMeta>)serviceLoadBalancerApi.select((List<ServiceInstance<ServiceMeta>>) serviceInstances);
        return serviceInstance.getPayload();
    }

    @Override
    public void init(RegisterConfig registerConfig) {
        CuratorFramework client = CuratorFrameworkFactory.newClient(registerConfig.getRegisterAddress(), new ExponentialBackoffRetry(BASE_SLEEP_TIME_MS, MAX_RETRIES));
        client.start();
        JsonInstanceSerializer<ServiceMeta> serializer = new JsonInstanceSerializer<>(ServiceMeta.class);
        this.serviceDiscovery = ServiceDiscoveryBuilder.builder(ServiceMeta.class)
                .client(client)
                .serializer(serializer)
                .basePath(ZK_BASE_PATH)
                .build();
        try {
            this.serviceDiscovery.start();
        } catch (Exception e) {
            LOGGER.error("serviceDiscovery start error :{}",e);
        }
    }
}
